package com.atguigu.edu.app.dwd.db;

import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderDetail {
    public static void main(String[] args) {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO　2 设置状态后端
                 /*
                 env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
                 env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
                 env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
                 env.setStateBackend(new HashMapStateBackend());
                 env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
                 System.setProperty("HADOOP_USER_NAME", "atguigu");
                  */

        //TODO 3 读取kafka的topic_db主题数据
        String topicName = "topic_db";
        String groupID = "dwd_trade_order_detail";
        tableEnv.executeSql("CREATE TABLE topic_db (\n" +
                "  `database` string,\n" +
                "  `table` string,\n" +
                "  `type` string,\n" +
                "  `ts` bigint,\n" +
                "  `xid` bigint,\n" +
                "  `commit` string,\n" +
                "  `data` map<string,string>,\n" +
                "  `old` map<string,string>, \n" +
                "   pt AS PROCTIME() \n" +
                ") " + KafkaUtil.getKafkaDDL(topicName, groupID));

        //TODO 4 过滤出订单表
        Table orderInfoTable = tableEnv.sqlQuery("select \n" +
                "  `data`['id'] id,\n" +
                "  `data`['order_status'] order_status,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['origin_amount'] origin_amount,\n" +
                "  `data`['coupon_reduce'] coupon_reduce,\n" +
                "  `data`['final_amount'] final_amount,\n" +
                "  `data`['update_time'] update_time,\n" +
                "  `data`['province_id'] province_id,\n" +
                "  `data`['session_id'] session_id,\n" +
                "  `ts`,\n" +
                "  `pt`\n" +
                "from topic_db\n" +
                "where`table`='order_info'\n" +
                "and `type`='insert' ");
        tableEnv.createTemporaryView("order_info",orderInfoTable);

        //TODO 5 过滤出订单详情表
        Table odTable = tableEnv.sqlQuery("select\n" +
                "`data`['id'] order_detail_id, \n" +
                "`data`['course_id'] course_id, \n" +
                "`data`['course_name'] course_name, \n" +
                "`data`['order_id'] order_id, \n" +
                "`data`['user_id'] user_id, \n" +
                "`data`['create_time'] create_time, \n" +
                "`ts`,\n" +
                "`pt`\n" +
                "from topic_db\n" +
                "where `table` = 'order_detail'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("order_detail",odTable);

        //TODO 6 关联订单表和订单详情表
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "oi.id,\n" +
                "oi.order_status,\n" +
                "oi.user_id,\n" +
                "oi.origin_amount,\n" +
                "oi.coupon_reduce,\n" +
                "oi.final_amount,\n" +
                "oi.update_time,\n" +
                "oi.province_id,\n" +
                "oi.session_id,\n" +
                "oi.`ts` oi_ts,\n" +
                "oi.`pt` pt,\n" +
                "od.order_detail_id,\n" +
                "od.course_id,\n" +
                "od.course_name,\n" +
                "od.create_time,\n" +
                "od.`ts` od_ts\n" +
                "from order_detail od\n" +
                "join order_info oi\n" +
                "on od.order_id = oi.id");
        tableEnv.createTemporaryView("result_table",resultTable);


        //TODO 7 写入到kafka新的主题dwd_trade_order_detail中
        tableEnv.executeSql("create table dwd_order_detail(\n" +
                "id string,\n" +
                "order_status string,\n" +
                "user_id string,\n" +
                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +
                "update_time string,\n" +
                "province_id string,\n" +
                "session_id string,\n" +
                "oi_ts bigint,\n" +
                "pt TIMESTAMP_LTZ(3),\n" +
                "order_detail_id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "create_time string,\n" +
                "od_ts bigint,\n" +
                "PRIMARY KEY (id) NOT ENFORCED\n" +
                ")" + KafkaUtil.getUpsertKafkaSinkDDL("dwd_trade_order_detail"));
        tableEnv.executeSql("insert into dwd_order_detail select * from result_table");
    }

}
